package org.onion.pattern.event.springboot.rocketmq.autoconfigure;


import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.consumer.EventConsumeInterceptor;
import org.onion.pattern.event.consumer.InterceptableEventConsumeHandler;
import org.onion.pattern.event.spring.io.EventCodecs;
import org.onion.pattern.event.springboot.rocketmq.annotation.RocketMQConsumerGroup;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQEventProperties;
import org.onion.pattern.event.springboot.rocketmq.consume.RocketMQMQEventConsumeHandler;
import org.onion.pattern.event.springboot.rocketmq.consume.RocketMQMessageConsumer;
import org.onion.pattern.event.springboot.rocketmq.consume.distribute.RocketEventListenerFactory;
import org.onion.pattern.event.springboot.rocketmq.consume.interceptor.RocketMQEventConsumeLogInterceptor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;


import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;


@Configuration(proxyBeanMethods = false)
@AutoConfigureAfter({RocketMQEventAutoConfigure.class})
//@ConditionalOnBean(value = RocketMQAutoConfiguration.class, annotation = EnableRocketMQEventListener.class)
@ConditionalOnProperty(prefix = "onion-pattern.event.rocketmq.consumer", value = "enabled", havingValue = "true", matchIfMissing = true)
@Slf4j
public class RocketMQConsumeAutoConfiguration implements ApplicationContextAware,
        ApplicationRunner, BeanPostProcessor  {

    @Autowired(required = false)
    private RocketMQEventProperties rocketMQEventConfig;
    @Autowired(required = false)
    private EventCodecs               eventCodecs;
    private GenericApplicationContext applicationContext;
    private volatile boolean                   inited = false;


    @Bean
    public RocketMQEventConsumeLogInterceptor rocketMQEventConsumeLogInterceptor(){
        return new RocketMQEventConsumeLogInterceptor();
    }

    @Bean
    public RocketEventListenerFactory rocketEventListenerFactory(){
        return new RocketEventListenerFactory();
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (inited || !(bean instanceof RocketMQEventProperties)) {
            return bean;
        }
        inited = true;

        rocketMQEventConfig = ObjectUtils.defaultIfNull(rocketMQEventConfig, this.applicationContext.getBean(RocketMQEventProperties.class));
        eventCodecs = ObjectUtils.defaultIfNull(eventCodecs, this.applicationContext.getBean(EventCodecs.class));

        //TODO 初始化消息接受者方式待优化，应该在 RocketMQGroup全部构建完后执行
        //限制rocketmq-client.log日志文件200MB大小
//        System.getProperty(ClientLogger.CLIENT_LOG_FILESIZE, String.valueOf(1024 * 200));
        final Map<String, Object> groupMap = this.applicationContext.getBeansWithAnnotation(RocketMQConsumerGroup.class);
        final Map<String, EventConsumeInterceptor> externalEventInterceptorsMap = applicationContext.getBeansOfType(EventConsumeInterceptor.class);
        final List<EventConsumeInterceptor> eventConsumeInterceptors = Lists.newArrayList(externalEventInterceptorsMap.values());
        eventConsumeInterceptors.sort(Comparator.comparingInt(EventConsumeInterceptor::order));

        for (Object rocketMQConsumerGroup : groupMap.values()) {
            this.initConsumerManager(rocketMQConsumerGroup, eventConsumeInterceptors);
        }
        return bean;
    }

    private void initConsumerManager(Object rocketMQConsumerGroup, List<EventConsumeInterceptor> interceptors) {
        EventConsumeHandler consumeHandler = new InterceptableEventConsumeHandler(
                new RocketMQMQEventConsumeHandler(rocketMQConsumerGroup, applicationContext),
                interceptors);

//        final DynamicConsumer dynamicConsumer = AopProxyUtils.ultimateTargetClass(rocketMQConsumerGroup)
//                .getAnnotation(DynamicConsumer.class);
//
        RocketMQMessageConsumer consumer = new RocketMQMessageConsumer(consumeHandler, rocketMQEventConfig, applicationContext);
        consumer.setEventCodecs(this.eventCodecs);
        applicationContext.registerBean(
                String.format("%s_%s", consumer.getClass().getSimpleName(), rocketMQConsumerGroup.getClass().getSimpleName()),
                RocketMQMessageConsumer.class, () -> consumer);
    }


    @Override
    public void run(ApplicationArguments args) {
        ForkJoinPool.commonPool().submit(this::createEventReceiver);
    }


    public void createEventReceiver() {
        try {
            RocketMQEventProperties properties = applicationContext
                    .getBean(RocketMQEventProperties.class);
            if (properties.getConsumer().isEnabled()) {
                final Map<String, RocketMQMessageConsumer> receiverMap = this.applicationContext.getBeansOfType(RocketMQMessageConsumer.class);
//        Runtime.getRuntime().addShutdownHook(new Thread(new ReceiveCloseRunnable(receiverMap.values())));
//                ExecutorService executorService = Executors.newFixedThreadPool(10);
                CompletableFuture<?> rootFeature = CompletableFuture.completedFuture(null);
                for (RocketMQMessageConsumer consumer : receiverMap.values()) {
                    CompletableFuture<Void> future = CompletableFuture.runAsync(consumer::start);
                    rootFeature = rootFeature.thenCompose(atomicInteger -> future.thenApply(val -> val));
                }
                rootFeature.get();
//                CompletableFuture.allOf(futureArr.toArray(new CompletableFuture[]{})).wait();
                log.info("【ROCKET事件订阅】订阅者全部处理成功");
            }
        } catch (Exception ex) {
            log.error("【ROCKET事件订阅】订阅者处理失败", ex);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (GenericApplicationContext) applicationContext;
    }


//    @Configuration(proxyBeanMethods = false)
//    public static class EmbeddedSubscribeHandlerConfiguration {
//
//        @Bean
//        @ConditionalOnBean(DomainEventConsumeLogger.class)
//        @ConditionalOnMissingBean
//        public ISubscribeHandler consumedMessageLoggerSubscribeHandler(
//                ObjectProvider<DomainEventConsumeLogger> logger) {
//            return new EmbeddedConsumedMessageLoggerSubscribeHandler(logger.getIfAvailable());
//        }
//
//        @Bean
//        @ConditionalOnBean({DomainEventIdempotentChecker.class, TransactionExecutor.class})
//        @ConditionalOnMissingBean
//        public ISubscribeHandler excatlyOnceSubscribeHandler(
//                ObjectProvider<DomainEventIdempotentChecker> supporter,
//                ObjectProvider<TransactionExecutor> executors) {
//            return new EmbeddedExcatlyOnceSubscribeHandler(supporter.getIfAvailable(),
//                    executors.getIfAvailable());
//        }
//
//        @Bean
////        @ConditionalOnBean(DomainEventConsumeLogger.class)
//        @ConditionalOnMissingBean
//        public ISubscribeHandler retryRuleSubscribeHandler() {
//            return new EmbeddedRetryRuleSubscribeHandler();
//        }
//
//    }

}
